package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo23Broadcast {
  // 广播变量
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo23Broadcast")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    // 黑名单
    val blackIds: Set[String] = Set[String]("1500100068", "1500100168", "1500100268", "1500100368", "1500100468")

    // 使用SparkContext将变量进行广播 得到一个广播变量
    val broBlackIds: Broadcast[Set[String]] = sc.broadcast(blackIds)

    // 基于黑名单数据 将黑名单中的学生信息过滤出来

    val stuRDD: RDD[String] = sc.textFile("Spark/data/students.txt")

    stuRDD
      .filter(line => {
        /**
         * 算子内部的代码会被封装成Task最终发送到Executor中去执行
         * 同理如果在算子内部使用了算子外部定义的集合
         * 那么同样会在每个Task中封装一份集合 如果分区数过多 则会造成Set集合的副本数量太多
         * Task最终是在Executor中执行的
         * Executor的数量 << Task的数量
         * 所以可以通过广播变量将算子外部定义的变量（基本数据类型、集合类型、自定义类都可）广播到每个Executor上
         * 使用广播变量可以大大减少副本数量，进而可以提高效率
         * Task中如果需要使用外部的变量时可以直接在Executor中获取
         */
        val id: String = line.split(",")(0)
        //        blackIds.contains(id)

        // 通过广播变量从Executor中获取数据
        val broSet: Set[String] = broBlackIds.value
        broSet.contains(id)
      })
      .foreach(println)


  }

}
